package com.zytech.zspider.client.eventlistener;

import com.zytech.modules.lang.Requires;
import com.zytech.zspider.core.TaskDefinition;
import com.zytech.zspider.core.TaskException;
import com.zytech.zspider.core.TaskResult;
import com.zytech.zspider.eventbus.Event;
import com.zytech.zspider.eventbus.EventBus;
import com.zytech.zspider.eventbus.EventEnvelope;
import com.zytech.zspider.eventbus.support.PipelineEventListener;
import com.zytech.zspider.logger.Loggers;
import com.zytech.zspider.task.TaskExecutor;
import com.zytech.zspider.task.event.TaskAssembledEvent;
import com.zytech.zspider.task.event.TaskEvents;

import java.util.List;

/**
 * @author siuming
 */
public class TaskEventListener extends PipelineEventListener {

    private final String nextQueueName;
    private final String failingQueueName;
    private final List<TaskExecutor> executors;

    /**
     * @param eventBus
     * @param nextQueueName
     * @param failingQueueName
     * @param executors
     */
    public TaskEventListener(EventBus eventBus,
                             String nextQueueName,
                             String failingQueueName,
                             List<TaskExecutor> executors) {

        super(eventBus);

        Requires.hasText(nextQueueName, "nextQueueName must not be null or empty.");
        Requires.hasText(failingQueueName, "failingQueueName must not be null or empty.");
        Requires.notNull(executors, "executors must not be null.");
        this.nextQueueName = nextQueueName;
        this.failingQueueName = failingQueueName;
        this.executors = executors;
    }

    @Override
    protected boolean beforeEvent(EventEnvelope envelope) {
        return envelope.getEvent() instanceof TaskAssembledEvent;
    }

    @Override
    protected EventTuple doOnEvent(EventEnvelope envelope) {
        Loggers.me().warn(getClass(), "receive new TaskAssembledEvent.");

        TaskAssembledEvent event = (TaskAssembledEvent) envelope.getEvent();
        TaskDefinition task = event.getTask();
        for (TaskExecutor executor : getExecutors()) {
            if (!executor.isTaskSupports(task)) {
                continue;
            }

            return submitTask(task, executor);
        }

        Loggers.me().warn(getClass(), "Not found TaskExecutor for the {}.", task);
        throw new TaskException("Not found TaskExecutor for the task.");
    }

    private EventTuple submitTask(TaskDefinition task, TaskExecutor executor) {
        TaskResult taskResult = executor.submitTask(task);
        Event nextEvent = TaskEvents.newTaskSucceed(task, taskResult);
        EventEnvelope nextEnvelope = EventEnvelope.of(nextEvent);
        return EventTuple.of(EventTuple.Type.QUEUE, getNextQueueName(), nextEnvelope);
    }

    @Override
    protected void handleEventException(EventEnvelope envelope, Exception e) {
        TaskDefinition task = ((TaskAssembledEvent) envelope.getEvent()).getTask();
        Event nextEvent = TaskEvents.newTaskFailed(task, e.getMessage());
        EventEnvelope newEnvelope = EventEnvelope.of(nextEvent);
        getEventBus().postQueue(getFailingQueueName(), newEnvelope);
    }

    /**
     * @return
     */
    protected final String getNextQueueName() {
        return nextQueueName;
    }

    /**
     * @return
     */
    protected final String getFailingQueueName() {
        return failingQueueName;
    }

    /**
     * @return
     */
    protected final List<TaskExecutor> getExecutors() {
        return executors;
    }
}
